package com.czk.shop.mq;

import com.alibaba.fastjson.JSON;
import com.comon.czk.constant.ShopCode;
import com.comon.czk.entity.MQEntity;
import com.czk.shop.mapper.TradeGoodsMapper;
import com.czk.shop.mapper.TradeGoodsNumberLogMapper;
import com.czk.shop.mapper.TradeMqConsumerLogMapper;
import com.czk.shop.pojo.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * 监听消息队列
 * 回退库存
 *
 * @Author:ChenZhangKun
 * @Date: 2020/10/20 21:23
 */
@Slf4j
@Component
@RocketMQMessageListener(topic = "${mq.order.topic}", consumerGroup = "${mq.order.consumer.group.name}", messageModel = MessageModel.BROADCASTING)
public class CancelMQListener implements RocketMQListener<MessageExt> {
    @Autowired
    private TradeMqConsumerLogMapper tradeMqConsumerLogMapper;
    @Value("${mq.order.consumer.group.name}")
    private String groupName;
    @Autowired
    private TradeGoodsMapper tradeGoodsMapper;
    @Autowired
    private TradeGoodsNumberLogMapper tradeGoodsNumberLogMapper;

    @Override
    public void onMessage(MessageExt messageExt) {
        String msgId = null;
        String tags = null;
        String keys = null;
        String body = null;
        try {
            // 1 解析消息内容
            msgId = messageExt.getMsgId();
            // tag
            tags = messageExt.getTags();
            // keys
            keys = messageExt.getKeys();
            // body
            body = new String(messageExt.getBody(), "UTF-8");
            // 2 查询消息消费记录
            TradeMqConsumerLogKey primaryKey = new TradeMqConsumerLogKey();
            primaryKey.setGroupName(groupName);
            primaryKey.setMsgKey(keys);
            primaryKey.setMsgTag(tags);
            TradeMqConsumerLog tradeMqConsumerLog = tradeMqConsumerLogMapper.selectByPrimaryKey(primaryKey);
            if (tradeMqConsumerLog != null) {
                // 消费过
                // 消息的处理状态
                Integer consumerStatus = tradeMqConsumerLog.getConsumerStatus();
                // 处理过
                if (ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode().intValue() == consumerStatus.intValue()) {
                    log.info("消息 " + msgId + " 已经处理过");
                    return;
                }
                if (ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode().intValue() == consumerStatus.intValue()) {
                    // 正在处理
                    log.info("消息 " + msgId + " 正在处理");
                    return;
                }
                if (ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode().intValue() == consumerStatus.intValue()) {
                    // 处理失败
                    // 获取处理失败次数
                    Integer consumerTimes = tradeMqConsumerLog.getConsumerTimes();
                    if (consumerTimes > 3) {
                        log.info("消息 " + msgId + " 处理次数超过3次，不能再处理了");
                        return;
                    }
                    // 数据库乐观锁
                    // 设置处理中
                    tradeMqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());
                    TradeMqConsumerLogExample example = new TradeMqConsumerLogExample();
                    TradeMqConsumerLogExample.Criteria criteria = example.createCriteria();
                    criteria.andMsgTagEqualTo(tags)
                            .andMsgKeyEqualTo(keys)
                            .andGroupNameEqualTo(groupName)
                            .andConsumerTimesEqualTo(tradeMqConsumerLog.getConsumerTimes());
                    // 数据库乐观锁处理，防止并发处理
                    int i = tradeMqConsumerLogMapper.updateByExampleSelective(tradeMqConsumerLog, example);
                    if (i <= 0) {
                        // 修改成功
                        log.info("并发修改，稍后处理");
                    }
                }
            } else {
                // 没有消费过
                tradeMqConsumerLog = new TradeMqConsumerLog();
                tradeMqConsumerLog.setMsgTag(tags);
                tradeMqConsumerLog.setMsgKey(keys);
                // 正在处理
                tradeMqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());
                tradeMqConsumerLog.setMsgId(msgId);
                tradeMqConsumerLog.setMsgBody(body);
                tradeMqConsumerLog.setConsumerTimes(0);
                // 插入数据库
                tradeMqConsumerLogMapper.insert(tradeMqConsumerLog);
            }
            // 3 回退库存
            MQEntity mqEntity = JSON.parseObject(body, MQEntity.class);
            Long goodsId = mqEntity.getGoodsId();
            // 拿到商品对象
            TradeGoods tradeGoods = tradeGoodsMapper.selectByPrimaryKey(goodsId);
            // 更改库存
            tradeGoods.setGoodsNumber(tradeGoods.getGoodsNumber() + mqEntity.getGoodsNum());
            // 更新操作
            tradeGoodsMapper.updateByPrimaryKey(tradeGoods);
            // 记录库存操作日志
            TradeGoodsNumberLog tradeGoodsNumberLog = new TradeGoodsNumberLog();
            tradeGoodsNumberLog.setOrderId(mqEntity.getOrderId());
            tradeGoodsNumberLog.setGoodsNumber(mqEntity.getGoodsNum());
            tradeGoodsNumberLog.setGoodsId(goodsId);
            tradeGoodsNumberLog.setLogTime(new Date());
            tradeGoodsNumberLogMapper.insert(tradeGoodsNumberLog);
            // 将消息的处理状态更改为成功
            tradeMqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode());
            tradeMqConsumerLog.setConsumerTimestamp(new Date());
            tradeMqConsumerLog.setRemark("回退成功");
            tradeMqConsumerLogMapper.updateByPrimaryKey(tradeMqConsumerLog);
            log.info("回退库存成功");
        } catch (Exception ex) {
            ex.printStackTrace();
            TradeMqConsumerLogKey primaryKey = new TradeMqConsumerLogKey();
            primaryKey.setGroupName(groupName);
            primaryKey.setMsgKey(keys);
            primaryKey.setMsgTag(tags);
            TradeMqConsumerLog tradeMqConsumerLog = tradeMqConsumerLogMapper.selectByPrimaryKey(primaryKey);
            if (tradeMqConsumerLog == null) {
                // 数据库没有记录
                // 没有消费过
                tradeMqConsumerLog = new TradeMqConsumerLog();
                tradeMqConsumerLog.setMsgTag(tags);
                tradeMqConsumerLog.setMsgKey(keys);
                // 正在处理
                tradeMqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode());
                tradeMqConsumerLog.setMsgId(msgId);
                tradeMqConsumerLog.setMsgBody(body);
                tradeMqConsumerLog.setConsumerTimes(1);
                // 插入
                tradeMqConsumerLogMapper.insert(tradeMqConsumerLog);
            } else {
                tradeMqConsumerLog.setConsumerTimes(tradeMqConsumerLog.getConsumerTimes() + 1);
                // 更新
                tradeMqConsumerLogMapper.updateByPrimaryKey(tradeMqConsumerLog);
            }
        }
    }
}
